package com.udf.flink.scala
//import org.apache.flink.api.common.typeinfo.TypeInformation
//import org.apache.flink.api.scala._
//import org.apache.flink.table.api._
//import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.TableEnvironment
import org.apache.flink.table.functions.{FunctionContext, ScalarFunction}

/**
 * Description: No Description
 *
 * @author tieyi  2019/12/11 13:57
 * @version 1.0.0
 */
object MyUDF {
  def main(args: Array[String]): Unit = {
//    val env = ExecutionEnvironment.getExecutionEnvironment
//    TableEnvironment env = TableEnvironment.create();
//    val tEnv = BatchTableEnvironment.create(env)
//    tEnv.registerFunction("StringLength", new StringLengUDF)

    // 环境配置// 环境配置

    val settings = EnvironmentSettings.newInstance.inStreamingMode.build

    val tEnv = TableEnvironment.create(settings)
    tEnv.createTemporaryFunction("StringLength", new StringLengUDF)
  }
}

class StringLengUDF extends ScalarFunction{

  override def open(context: FunctionContext): Unit = super.open(context)

  override def close(): Unit = super.close()

  def eval(s: String): Int = {
    if (s == null){
      0
    }else{
      s.length
    }
  }
}